package com.tdy.cdc.app.dwd;

import com.tdy.cdc.app.BaseSqlApp;
import com.tdy.cdc.common.Constant;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.FlinkSourceUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamStatementSet;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: NanHuang
 * @Date: 2023/04/03/10:38
 * @Description:
 */
public class App_05_08 extends BaseSqlApp {
    public static void main(String[] args) {
        new App_05_08().init(
                4002,
                "App_05_08",
                2
        );
    }
    @Override
    protected void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 创建源表
        createSourceTables(tableEnv);
        // 2 join
        joinTables(tableEnv);
        // 3 写入Kafka
        writeToKafka(tableEnv);
    }

    private void writeToKafka(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table target5 (" +
                "XDS_TYPE                       STRING, " +
                "XDS_NAME                       STRING, " +
                "XDS_VERSION                    STRING, " +
                "PATIENT_DOMAIN                 STRING, " +
                "VISIT_DOMAIN                   STRING, " +
                "CUSTODIAN_CODE                 STRING, " +
                "CUSTODIAN_NAME                 STRING, " +
                "HOSPITAL_CODE                  STRING, " +
                "HOSPITAL_NAME                  STRING, " +
                "PATIENT_ID                     STRING, " +
                "VISIT_ID                       STRING, " +
                "INPATIENT_VISIT_FLOW_ID        STRING, " +
                "OUTPATIENT_VISIT_FLOW_ID       STRING, " +
                "VISIT_TIMES                    STRING, " +
                "RECORD_ID                      STRING, " +
                "DW_DATE                        STRING, " +
                "OUTPATIENT_NO                  STRING, " +
                "DOMAIN_ID                      STRING, " +
                "VISIT_DATE                     STRING, " +
                "INPATIENT_NO                   STRING, " +
                "PATIENT_VISIT_CATEGORY_CODE    STRING, " +
                "PATIENT_VISIT_CATEGORY         STRING, " +
                "TEST_REPORT_NO                 STRING, " +
                "APPLY_SHEET_NO                 STRING, " +
                "SPECIMEN_NUMBER                STRING, " +
                "TELECOM                        STRING, " +
                "ID_CARD_NUMBER                 STRING, " +
                "PATIENT_NAME                   STRING, " +
                "SEX_CODE                       STRING, " +
                "SEX_NAME                       STRING, " +
                "AGE                            STRING, " +
                "AGE_UNIT                       STRING, " +
                "TEST_DATE                      STRING, " +
                "TEST_REPORT_DOCTOR_ID          STRING, " +
                "TEST_REPORT_DOCTOR_NAME        STRING, " +
                "AUDIT_SIGNDATE                 STRING, " +
                "TEST_AUDIT_DOCTOR_ID           STRING, " +
                "TEST_AUDIT_DOCTOR_NAME         STRING, " +
                "CHECK_SIGNDATE                 STRING, " +
                "TEST_TECH_STAFF_ID             STRING, " +
                "TEST_TECH_STAFF_NAME           STRING, " +
                "TECHNICIAN_SIGNDATE            STRING, " +
                "TEST_DOCTOR_ID                 STRING, " +
                "TEST_DOCTOR_NAME               STRING, " +
                "APPLY_DATE                     STRING, " +
                "APPLY_DEPT_ID                  STRING, " +
                "APPLY_DEPT_NAME                STRING, " +
                "APPLY_MED_ORG_ID               STRING, " +
                "TEST_REPORT_DEPT_ID            STRING, " +
                "BED_NO                         STRING, " +
                "BED_NAME                       STRING, " +
                "ROOM_NO                        STRING, " +
                "ROOM_NAME                      STRING, " +
                "DEPT_ID                        STRING, " +
                "DEPT_NAME                      STRING, " +
                "WARDS_ID                       STRING, " +
                "WARDS_NAME                     STRING, " +
                "INSPECTION_METHOD              STRING, " +
                "INSPECT_TYPE                   STRING, " +
                "TEST_REPORT_RESULT             STRING, " +
                "REPORT_DEPT                    STRING, " +
                "AGENCY_NAME                    STRING, " +
                "TEST_REPORT_COOMMENTS          STRING, " +
                "INSPECT_TYPE_CODE              STRING, " +
                "INSPECTION_ITEM                STRING, " +
                "EXAM_DATE                      STRING, " +
                "ITEM_CODE                      STRING, " +
                "ITEM_NAME                      STRING, " +
                "SPECIMEN_SAMPLE_TIME           STRING, " +
                "SPECIMEN_RECV_TIME             STRING, " +
                "SPECIMEN_CLASS                 STRING, " +
                "SPECIMEN_STATUS                STRING, " +
                "SPECIMEN_FIXATIVE_NAME         STRING, " +
                "RESULT_NAME                    STRING, " +
                "RESULT_CODE                    STRING, " +
                "EXAM_RESULT_VALUE              STRING, " +
                "EXAM_RESULT_UNIT               STRING, " +
                "DIAG_DATE                      STRING, " +
                "DIAG_CODE                      STRING, " +
                "DIAG_NAME                      STRING, " +
                "DIAG_TYPE                      STRING, " +
                "DIAG_INSTITUTION               STRING, " +
                "primary key (RECORD_ID) not enforced" +
                ")" +
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_TEST_REPORT));

        tableEnv.executeSql("create table target6 (" +
                "RECORD_ID                          STRING," +
                "MED_ORG_NAME                       STRING," +
                "MED_ORG_CODE                       STRING," +
                "HOSPITAL_NAME                      STRING," +
                "HOSPITAL_CODE                      STRING," +
                "AGE_UNIT                           STRING," +
                "AGE                                STRING," +
                "SEX_NAME                           STRING," +
                "SEX_CODE                           STRING," +
                "ID_CARD_NUMBER                     STRING," +
                "PATIENT_VISIT_CATEGORY_CODE        STRING," +
                "PATIENT_VISIT_CATEGORY             STRING," +
                "INPATIENT_VISIT_FLOW_ID            STRING," +
                "INPATIENT_NO                       STRING," +
                "OUTPATIENT_VISIT_FLOW_ID           STRING," +
                "OUTPATIENT_NO                      STRING," +
                "PATIENT_ID                         STRING," +
                "PATIENT_NAME                       STRING," +
                "DOMAIN_ID                          STRING," +
                "BED_NAME                           STRING," +
                "VISIT_TIMES                        STRING," +
                "APPLY_SHEET_NO                     STRING," +
                "AUTHOR_CODE                        STRING," +
                "AUTHOR_NAME                        STRING," +
                "CUSTODIAN_CODE                     STRING," +
                "CUSTODIAN_NAME                     STRING," +
                "EXECUTOR_DATE                      STRING," +
                "ORDERS_EXEC_STAFF_ID               STRING," +
                "ORDERS_EXEC_STAFF_NAME             STRING," +
                "VISIT_DATE                         STRING," +
                "BED_NO                             STRING," +
                "DEPT_ID                            STRING," +
                "DEPT_NAME                          STRING," +
                "WARD_ID                            STRING," +
                "WARD_NAME                          STRING," +
                "INVASIVE_TREAT_FLAG                STRING," +
                "ALLERGIC_HISTORY_FLAG              STRING," +
                "ALLERGIC_HISTORY                   STRING," +
                "PATIENT_WEIGHT                     STRING," +
                "DISPOSE_INSTRUCTION                STRING," +
                "ORDERS_USE_COMMENTS                STRING," +
                "NEXT_TREAT_SCHEME                  STRING," +
                "FOLLOW_UP_DATE                     STRING," +
                "FOLLOW_UP_WAY_CODE                 STRING," +
                "FOLLOW_UP_WAY_NAME                 STRING," +
                "FOLLOW_UP_CYCLE_CODE               STRING," +
                "FOLLOW_UP_CYCLE_NAME               STRING," +
                "ROOM_NO                            STRING," +
                "ROOM_NAME                          STRING," +
                "DIAG_CODE                          STRING," +
                "DIAG_NAME                          STRING," +
                "XDS_TYPE                           STRING," +
                "XDS_NAME                           STRING," +
                "XDS_VERSION                        STRING," +
                "VISIT_ID                           STRING," +
                "table_name                         STRING," +
                "primary key (RECORD_ID) not enforced" +
                ")" +
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_THERAPY_RECORD));

        tableEnv.executeSql("create table target7 (" +
                "RECORD_ID                     String,"+
                "MED_ORG_NAME                  String,"+
                "HOSPITAL_NAME                 String,"+
                "HOSPITAL_CODE                 String,"+
                "AGE_UNIT                      String,"+
                "AGE                           String,"+
                "SEX_NAME                      String,"+
                "SEX_CODE                      String,"+
                "ID_CARD_NUMBER                String,"+
                "PATIENT_VISIT_CATEGORY_CODE   String,"+
                "PATIENT_VISIT_CATEGORY        String,"+
                "INPATIENT_VISIT_FLOW_ID       String,"+
                "INPATIENT_NO                  String,"+
                "OUTPATIENT_VISIT_FLOW_ID      String,"+
                "OUTPATIENT_NO                 String,"+
                "PATIENT_ID                    String,"+
                "PATIENT_NAME                  String,"+
                "DOMAIN_ID                     String,"+
                "WARD_ID                       String,"+
                "WARD_NAME                     String,"+
                "VISIT_TIMES                   String,"+
                "APPLY_SHEET_NO                String,"+
                "AUTHOR_CODE                   String,"+
                "AUTHOR_NAME                   String,"+
                "CUSTODIAN_CODE                String,"+
                "CUSTODIAN_NAME                String,"+
                "SURGEON_SIGNDATE              String,"+
                "VISIT_DATE                    String,"+
                "BED_NO                        String,"+
                "BED_NAME                      String,"+
                "DEPT_ID                       String,"+
                "DEPT_NAME                     String,"+
                "SURGERY_HISTORY               String,"+
                "OPER_CODE                     String,"+
                "OPER_NAME                     String,"+
                "OPER_LEVEL_CODE               String,"+
                "OPER_LEVEL                    String,"+
                "OPER_START_DATE               String,"+
                "OPER_END_DATE                 String,"+
                "FIRST_ASSISTANT_ID            String,"+
                "FIRST_ASSISTANT_NAME          String,"+
                "SECOND_ASSISTANT_ID           String,"+
                "SECOND_ASSISTANT_NAME         String,"+
                "INSTRUMENT_NURSE_ID           String,"+
                "INSTRUMENT_NURSE_NAME         String,"+
                "VISITATORIAL_NURSE_ID         String,"+
                "VISITATORIAL_NURSE_NAME       String,"+
                "OPER_ROOM                     String,"+
                "BLEEDING                      String,"+
                "METACHYSIS_VOLUME             String,"+
                "METACHYSIS_REACTION_FLAG      String,"+
                "ANES_METHOD_CODE              String,"+
                "ANES_METHOD_NAME              String,"+
                "PRE_MEDICATION                String,"+
                "INOPER_MEDICAL                String,"+
                "INFUSION_VOLUME               String,"+
                "OPERATION_DESC                String,"+
                "TARGET_NAME                   String,"+
                "INTERVENTION                  String,"+
                "OPER_POST_CODE                String,"+
                "OPER_POST                     String,"+
                "SKIN_DISINFECTION             String,"+
                "INCISION_DESC                 String,"+
                "DRAINAGE_MARK                 String,"+
                "DRAINAGE_MATERIAL             String,"+
                "DRAINAGE_MATERIAL_NUM         String,"+
                "PLACEMENT_SITE                String,"+
                "ROOM_NAME                     String,"+
                "ANES_DOCTOR_NAME              String,"+
                "ROOM_NO                       String,"+
                "PRE_OPER_DIAG_ICD_NAME        String,"+
                "PRE_OPER_DIAG_ICD_CODE        String,"+
                "POST_OPER_DIAG_ICD_CODE       String,"+
                "POST_OPER_DIAG_ICD_NAME       String," +
                "XDS_TYPE                      STRING," +
                "XDS_NAME                      STRING," +
                "XDS_VERSION                   STRING," +
                "VISIT_ID                      STRING,"+
                "table_name                    STRING," +
                "primary key (RECORD_ID) not enforced" +
                ")" +
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_GENERAL_SURGERY_RECORD));

        tableEnv.executeSql("create table target8 (" +
                "RECORD_ID                      String,"+
                "MED_ORG_NAME                   String,"+
                "MED_ORG_CODE                   String,"+
                "HOSPITAL_NAME                  String,"+
                "HOSPITAL_CODE                  String,"+
                "AGE_UNIT                       String,"+
                "AGE                            String,"+
                "SEX_NAME                       String,"+
                "SEX_CODE                       String,"+
                "ID_CARD_NUMBER                 String,"+
                "PATIENT_VISIT_CATEGORY_CODE    String,"+
                "PATIENT_VISIT_CATEGORY         String,"+
                "INPATIENT_VISIT_FLOW_ID        String,"+
                "INPATIENT_NO                   String,"+
                "OUTPATIENT_VISIT_FLOW_ID       String,"+
                "OUTPATIENT_NO                  String,"+
                "PATIENT_ID                     String,"+
                "PATIENT_NAME                   String,"+
                "DOMAIN_ID                      String,"+
                "BED_NO                         String,"+
                "BED_NAME                       String,"+
                "WARD_ID                        String,"+
                "WARD_NAME                      String,"+
                "VISIT_DATE                     String,"+
                "VISIT_TIMES                    String,"+
                "APPY_SHEET_NO                  String,"+
                "AUTHOR_CODE                    String,"+
                "AUTHOR_NAME                    String,"+
                "CUSTODIAN_CODE                 String,"+
                "CUSTODIAN_NAME                 String,"+
                "ANESTH_SIGNDATE                String,"+
                "ANESTH_CODE                    String,"+
                "ANES_DOCTOR_NAME               String,"+
                "ADMIT_DATETIME                 String,"+
                "DEPT_ID                        String,"+
                "DEPT_NAME                      String,"+
                "PREOPERATIVE_COMPLICATION      String,"+
                "BRIEF_MEDICAL_HISTORY          String,"+
                "ALLERGY_HISTORY                String,"+
                "PATIENT_WEIGHT                 String,"+
                "MENTAL_STATE_FLAG              String,"+
                "ABO_BLOOD_CODE                 String,"+
                "ABO_BLOOD_NAME                 String,"+
                "RH_BLOOD_CODE                  String,"+
                "RH_BLOOD_NAME                  String,"+
                "OPERATE_ROOM_NO                String,"+
                "PRE_OPER_DIAG_ICD_CODE         String,"+
                "PRE_OPER_DIAG_ICD_NAME         String,"+
                "CONSIDERATIONS                 String,"+
                "CARDIOGRAM_EXAM_RESULT         String,"+
                "CHEST_RAY_EXAM_RESULT          String,"+
                "CT_EXAM_RESULT                 String,"+
                "B_ULTRAPHONIC_EXAM_RESULT      String,"+
                "MRI_EXAM_RESULT                String,"+
                "LUNG_FUNCTION_EXAM_RESULT      String,"+
                "BLOOD_ROUTINE_EXAM_RESULT      String,"+
                "URINE_ROUTINE_EXAM_RESUTL      String,"+
                "COAGULATION_EXAM_RESULT        String,"+
                "LIVER_FUNCTION_EXAM_RESULT     String,"+
                "BLOOD_GAS_ANALYSIS_RESULT      String,"+
                "GENERAL_EXAM_RESULT            String,"+
                "HEART_AUSCULTATE_RESULT        String,"+
                "LUNG_AUSCULTATE_RESULT         String,"+
                "ROOM_NO                        String,"+
                "ROOM_NAME                      String,"+
                "LIMBS_EXAM_RESULT              String,"+
                "SPINE_EXAM_RESULT              String,"+
                "STOMACH_EXAM_RESULT            String,"+
                "TRACHEA_EXAM_RESULT            String,"+
                "TOOTH_EXAM_RESULT              String,"+
                "ANES_INDICATION                String,"+
                "ANES_DOCTOR_ID                 String,"+
                "ANES_DOCTOR_SIGN_NAME          String,"+
                "ELEC_SIGN_DATETIME             String,"+
                "INTEND_ANES_METHOD_CODE        String,"+
                "INTEND_ANES_METHOD_NAME        String,"+
                "OPER_CODE                      String,"+
                "OPER_NAME                      String," +
                "XDS_TYPE                       STRING," +
                "XDS_NAME                       STRING," +
                "XDS_VERSION                    STRING," +
                "VISIT_ID                       String,"+
                "table_name                     STRING," +
                "primary key (RECORD_ID) not enforced" +
                ")" +
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_PREOPERATIVE_VISIT));

        // 导入数据
        StreamStatementSet statementSet = tableEnv.createStatementSet();
        statementSet.addInsertSql("insert into target5 select * from join_result5")
                .addInsertSql("insert into target6 select * from join_result6")
                .addInsertSql("insert into target7 select * from join_result7")
                .addInsertSql("insert into target8 select * from join_result8");
        statementSet.execute();
    }

    private void joinTables(StreamTableEnvironment tableEnv) {
        // 1 设置数据的TTL（暂时30min）
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60));
        // 2 join操作
        Table joinResult5 = tableEnv.sqlQuery("select " +
                "  s.xds_type              as   XDS_TYPE            ,  " +
                "  s.xds_name              as   XDS_NAME            ,  " +
                "  s.xds_version           as   XDS_VERSION         ,  " +
                "  s.patient_domain        as   PATIENT_DOMAIN      ,  " +
                "  s.visit_domain          as   VISIT_DOMAIN        ,  " +
                "  s.custodian_code        as   CUSTODIAN_CODE      ,  " +
                "  s.custodian_name        as   CUSTODIAN_NAME      ,  " +
                "  s.hospital_code         as   HOSPITAL_CODE       ,  " +
                "  s.hospital_name         as   HOSPITAL_NAME       ,  " +
                "  s.patient_id            as   PATIENT_ID          ,  " +
                "  s.visit_id              as   VISIT_ID            ,  " +
                "  case when s.patient_typecode = '3'  then visit_id else '' end as INPATIENT_VISIT_FLOW_ID        ,  " +
                "    case when s.patient_typecode <> '3' then visit_id else '' end as OUTPATIENT_VISIT_FLOW_ID       ,  " +
                "  s.visit_times           as   VISIT_TIMES                       ,  " +
                "  s.unique_id             as   RECORD_ID                         ,  " +
                "  s.effective_time        as   DW_DATE                           ,  " +
                "  s.out_patient_id        as   OUTPATIENT_NO                     ,  " +
                "  s.domain_id             as   DOMAIN_ID                         ,  " +
                "  s.visit_date            as   VISIT_DATE                        ,  " +
                "  s.inpatient             as   INPATIENT_NO                      ,  " +
                "  s.patient_typecode      as   PATIENT_VISIT_CATEGORY_CODE       ,  " +
                "  s.patient_type          as   PATIENT_VISIT_CATEGORY            ,  " +
                "  s.report_id             as   TEST_REPORT_NO                    ,  " +
                "  s.appy_id               as   APPLY_SHEET_NO                    ,  " +
                "  s.specimen_number       as   SPECIMEN_NUMBER                   ,  " +
                "  s.telecom               as   TELECOM                           ,  " +
                "  s.identify_no           as   ID_CARD_NUMBER                    ,  " +
                "  s.patient_name          as   PATIENT_NAME                      ,  " +
                "  s.sex_code              as   SEX_CODE                          ,  " +
                "  s.sex                   as   SEX_NAME                          ,  " +
                "  s.age                   as   AGE                               ,  " +
                "  s.age_unit              as   AGE_UNIT                          ,  " +
                "  s.write_time            as   TEST_DATE                         ,  " +
                "  s.author_code           as   TEST_REPORT_DOCTOR_ID             ,  " +
                "  s.author_name           as   TEST_REPORT_DOCTOR_NAME           ,  " +
                "  s.audit_signdate        as   AUDIT_SIGNDATE                    ,  " +
                "  s.audit_code            as   TEST_AUDIT_DOCTOR_ID              ,  " +
                "  s.audit_name            as   TEST_AUDIT_DOCTOR_NAME            ,  " +
                "  s.check_signdate        as   CHECK_SIGNDATE                    ,  " +
                "  s.technician_code       as   TEST_TECH_STAFF_ID                ,  " +
                "  s.technician_name       as   TEST_TECH_STAFF_NAME              ,  " +
                "  s.technician_signdate   as   TECHNICIAN_SIGNDATE               ,  " +
                "  s.examiner_code         as   TEST_DOCTOR_ID                    ,  " +
                "  s.examiner_name         as   TEST_DOCTOR_NAME                  ,  " +
                "  s.apply_date            as   APPLY_DATE                        ,  " +
                "  s.apply_dept_code       as   APPLY_DEPT_ID                     ,  " +
                "  s.apply_dept            as   APPLY_DEPT_NAME                   ,  " +
                "  s.apply_agencycode      as   APPLY_MED_ORG_ID                  ,  " +
                "  s.apply_organ           as   TEST_REPORT_DEPT_ID               ,  " +
                "  s.bed_no                as   BED_NO                            ,  " +
                "  s.bed_name              as   BED_NAME                          ,  " +
                "  s.ward_id               as   ROOM_NO                           ,  " +
                "  s.ward_name             as   ROOM_NAME                         ,  " +
                "  s.dept_code             as   DEPT_ID                           ,  " +
                "  s.dept_name             as   DEPT_NAME                         ,  " +
                "  s.wards_id              as   WARDS_ID                          ,  " +
                "  s.wards_name            as   WARDS_NAME                        ,  " +
                "  s.inspection_method     as   INSPECTION_METHOD                 ,  " +
                "  s.inspect_type          as   INSPECT_TYPE                      ,  " +
                "  s.report_result         as   TEST_REPORT_RESULT                ,  " +
                "  s.report_dept           as   REPORT_DEPT                       ,  " +
                "  s.agency_name           as   AGENCY_NAME                       ,  " +
                "  s.report_remark         as   TEST_REPORT_COOMMENTS             ,  " +
                "  s.inspect_type_code     as   INSPECT_TYPE_CODE                 ,  " +
                "  s.inspection_item       as   INSPECTION_ITEM                   ,  " +
                "  e.check_date            as  EXAM_DATE                          ,  " +
                "  e.item_code             as  ITEM_CODE                          ,  " +
                "  e.item_name             as  ITEM_NAME                          ,  " +
                "  e.sample_date           as  SPECIMEN_SAMPLE_TIME               ,  " +
                "  e.receipt_date          as  SPECIMEN_RECV_TIME                 ,  " +
                "  e.specimen_category     as  SPECIMEN_CLASS                     ,  " +
                "  e.specimen_status       as  SPECIMEN_STATUS                    ,  " +
                "  e.specimen_fixative     as  SPECIMEN_FIXATIVE_NAME             ,  " +
                "  e.result_name           as  RESULT_NAME                        ,  " +
                "  e.result_code           as  RESULT_CODE                        ,  " +
                "  e.quanty_result         as  EXAM_RESULT_VALUE                  ,  " +
                "  e.quanty_unit           as  EXAM_RESULT_UNIT                   ,  " +
                "  d.diag_date             as  DIAG_DATE                          ,  " +
                "  d.diag_code             as  DIAG_CODE                          ,  " +
                "  d.diag_name             as  DIAG_NAME                          ,  " +
                "  d.diag_type             as  DIAG_TYPE                          ,  " +
                "  d.diag_institution      as  DIAG_INSTITUTION                      " +
                "from hdsd00_05_01 s  " +
                "left join tdy_list_diag d on s.unique_id = d.unique_id and s.xds_type = d.xds_type  " +
                "left join tdy_list_exam e on s.unique_id = e.unique_id and s.xds_type = e.xds_type");
        tableEnv.createTemporaryView("join_result5",joinResult5);

        Table joinResult6 = tableEnv.sqlQuery("select " +
                "s.unique_id as RECORD_ID , " +
                "hospital_name as MED_ORG_NAME , " +
                "hospital_code as MED_ORG_CODE , " +
                "hospital_name as HOSPITAL_NAME , " +
                "hospital_code as HOSPITAL_CODE , " +
                "age_unit as AGE_UNIT , " +
                "age as AGE , " +
                "sex as SEX_NAME , " +
                "sex_code as SEX_CODE , " +
                "identify_no as ID_CARD_NUMBER , " +
                "patient_typecode as PATIENT_VISIT_CATEGORY_CODE , " +
                "patient_type as PATIENT_VISIT_CATEGORY , " +
                "case when patient_typecode = '3' then visit_id else '' end as INPATIENT_VISIT_FLOW_ID , " +
                "inpatient as INPATIENT_NO , " +
                "case when patient_typecode <> '3' then visit_id else '' end as OUTPATIENT_VISIT_FLOW_ID , " +
                "out_patient_id as OUTPATIENT_NO , " +
                "patient_id as PATIENT_ID , " +
                "patient_name as PATIENT_NAME , " +
                "s.domain_id as DOMAIN_ID , " +
                "bed_name as BED_NAME , " +
                "visit_times as VISIT_TIMES , " +
                "appy_id as APPLY_SHEET_NO , " +
                "author_code as AUTHOR_CODE , " +
                "author_name as AUTHOR_NAME , " +
                "custodian_code as CUSTODIAN_CODE , " +
                "custodian_name as CUSTODIAN_NAME , " +
                "executor_date as EXECUTOR_DATE , " +
                "executor_code as ORDERS_EXEC_STAFF_ID , " +
                "executor_order as ORDERS_EXEC_STAFF_NAME , " +
                "visit_date as VISIT_DATE , " +
                "bed_no as BED_NO , " +
                "dept_code as DEPT_ID , " +
                "dept_name as DEPT_NAME , " +
                "wards_id as WARD_ID , " +
                "wards_name as WARD_NAME , " +
                "invasive_sign as INVASIVE_TREAT_FLAG , " +
                "allergy_history_mark as ALLERGIC_HISTORY_FLAG , " +
                "allergy_history as ALLERGIC_HISTORY , " +
                "weight as PATIENT_WEIGHT , " +
                "handling_guidance as DISPOSE_INSTRUCTION , " +
                "order_use_remark as ORDERS_USE_COMMENTS , " +
                "treatment_plan as NEXT_TREAT_SCHEME , " +
                "followup_date as FOLLOW_UP_DATE , " +
                "fu_way_code as FOLLOW_UP_WAY_CODE , " +
                "fu_way as FOLLOW_UP_WAY_NAME , " +
                "followup_recommendcode as FOLLOW_UP_CYCLE_CODE , " +
                "followup_recommend as FOLLOW_UP_CYCLE_NAME , " +
                "ward_id as ROOM_NO , " +
                "ward_name as ROOM_NAME , " +
                "diag_code as DIAG_CODE , " +
                "diag_name as DIAG_NAME ," +
                "s.xds_type as XDS_TYPE," +
                "s.xds_name as XDS_NAME," +
                "s.xds_version as XDS_VERSION," +
                "visit_id as VISIT_ID," +
                "'dwd_mhs_therapy_record' as table_name " +
                "from hdsd00_06_03 s    " +
                "left join tdy_list_diag d on s.unique_id = d.unique_id and s.xds_type = d.xds_type    ");
        tableEnv.createTemporaryView("join_result6",joinResult6);

        Table joinResult7 = tableEnv.sqlQuery("select     " +
                "s.unique_id                                                     AS   RECORD_ID   ,"+
                "hospital_name                                                   AS   MED_ORG_NAME   ,"+
                "hospital_name                                                   AS   HOSPITAL_NAME   ,"+
                "hospital_code                                                   AS   HOSPITAL_CODE   ,"+
                "age_unit                                                        AS   AGE_UNIT   ,"+
                "age                                                             AS   AGE   ,"+
                "sex                                                             AS   SEX_NAME   ,"+
                "sex_code                                                        AS   SEX_CODE   ,"+
                "identify_no                                                     AS   ID_CARD_NUMBER   ,"+
                "patient_typecode                                                AS   PATIENT_VISIT_CATEGORY_CODE   ,"+
                "patient_type                                                    AS   PATIENT_VISIT_CATEGORY   ,"+
                "case when patient_typecode ='3' then visit_id else '' end       AS   INPATIENT_VISIT_FLOW_ID   ,"+
                "inpatient                                                       AS   INPATIENT_NO   ,"+
                "case when patient_typecode <> '3' then visit_id else '' end     AS   OUTPATIENT_VISIT_FLOW_ID   ,"+
                "out_patient_id                                                  AS   OUTPATIENT_NO   ,"+
                "patient_id                                                      AS   PATIENT_ID   ,"+
                "patient_name                                                    AS   PATIENT_NAME   ,"+
                "s.domain_id                                                     AS   DOMAIN_ID   ,"+
                "wards_id                                                        AS   WARD_ID   ,"+
                "wards_name                                                      AS   WARD_NAME   ,"+
                "visit_times                                                     AS   VISIT_TIMES   ,"+
                "appy_id                                                         AS   APPLY_SHEET_NO   ,"+
                "author_code                                                     AS   AUTHOR_CODE   ,"+
                "author_name                                                     AS   AUTHOR_NAME   ,"+
                "custodian_code                                                  AS   CUSTODIAN_CODE   ,"+
                "custodian_name                                                  AS   CUSTODIAN_NAME   ,"+
                "surgeon_signdate                                                AS   SURGEON_SIGNDATE   ,"+
                "visit_date                                                      AS   VISIT_DATE   ,"+
                "bed_no                                                          AS   BED_NO   ,"+
                "bed_name                                                        AS   BED_NAME   ,"+
                "dept_code                                                       AS   DEPT_ID   ,"+
                "dept_name                                                       AS   DEPT_NAME   ,"+
                "surgery_history                                                 AS   SURGERY_HISTORY   ,"+
                "oper_code                                                       AS   OPER_CODE   ,"+
                "oper_name                                                       AS   OPER_NAME   ,"+
                "oper_level_code                                                 AS   OPER_LEVEL_CODE   ,"+
                "oper_level                                                      AS   OPER_LEVEL   ,"+
                "oper_start_date                                                 AS   OPER_START_DATE   ,"+
                "oper_end_date                                                   AS   OPER_END_DATE   ,"+
                "assistant_icode                                                 AS   FIRST_ASSISTANT_ID   ,"+
                "assistant_i                                                     AS   FIRST_ASSISTANT_NAME   ,"+
                "assistant_iicode                                                AS   SECOND_ASSISTANT_ID   ,"+
                "assistant_ii                                                    AS   SECOND_ASSISTANT_NAME   ,"+
                "instrument_code                                                 AS   INSTRUMENT_NURSE_ID   ,"+
                "instrument_name                                                 AS   INSTRUMENT_NURSE_NAME   ,"+
                "patrol_code                                                     AS   VISITATORIAL_NURSE_ID   ,"+
                "patrol_name                                                     AS   VISITATORIAL_NURSE_NAME   ,"+
                "oper_room                                                       AS   OPER_ROOM   ,"+
                "bleeding                                                        AS   BLEEDING   ,"+
                "blood_volume                                                    AS   METACHYSIS_VOLUME   ,"+
                "cast(case when blood_reaction_mark='1' then blood_reaction_mark='true' else blood_reaction_mark='false' end as string)  AS   METACHYSIS_REACTION_FLAG   ,"+
                "anes_method_code                                                AS   ANES_METHOD_CODE   ,"+
                "anes_method_name                                                AS   ANES_METHOD_NAME   ,"+
                "pre_medication                                                  AS   PRE_MEDICATION   ,"+
                "inoper_medical                                                  AS   INOPER_MEDICAL   ,"+
                "infusion_volume                                                 AS   INFUSION_VOLUME   ,"+
                "operation_desc                                                  AS   OPERATION_DESC   ,"+
                "target_name                                                     AS   TARGET_NAME   ,"+
                "intervention                                                    AS   INTERVENTION   ,"+
                "oper_post_code                                                  AS   OPER_POST_CODE   ,"+
                "oper_post                                                       AS   OPER_POST   ,"+
                "skin_disinfection                                               AS   SKIN_DISINFECTION   ,"+
                "incision_desc                                                   AS   INCISION_DESC   ,"+
                "cast(case when drainage_mark ='1' then drainage_mark ='true' else drainage_mark ='false' end as string)  AS   DRAINAGE_MARK   ,"+
                "drainage_material                                               AS   DRAINAGE_MATERIAL   ,"+
                "drainage_material_num                                           AS   DRAINAGE_MATERIAL_NUM   ,"+
                "placement_site                                                  AS   PLACEMENT_SITE   ,"+
                "ward_name                                                       AS   ROOM_NAME      ,"+
                "anesth_name                                                     AS   ANES_DOCTOR_NAME      ,"+
                "ward_id                                                         AS   ROOM_NO      ,"+
                "case when diag_type ='术前诊断' then diag_code else '' end       AS   PRE_OPER_DIAG_ICD_CODE     ,"+
                "case when diag_type ='术前诊断' then diag_name else '' end       AS   PRE_OPER_DIAG_ICD_NAME     ,"+
                "case when diag_type ='术后诊断' then diag_code else '' end       AS   POST_OPER_DIAG_ICD_CODE     ,"+
                "case when diag_type ='术后诊断' then diag_name else '' end       AS   POST_OPER_DIAG_ICD_NAME ," +
                "s.xds_type as XDS_TYPE," +
                "s.xds_name as XDS_NAME," +
                "s.xds_version as XDS_VERSION," +
                "visit_id                                                        AS   VISIT_ID," +
                "'dwd_mhs_general_surgery_record' as table_name  "+
                "from hdsd00_06_06 s " +
                "left join tdy_list_diag a " +
                "on a.unique_id=s.unique_id and a.xds_type=s.xds_type"
        );
        tableEnv.createTemporaryView("join_result7",joinResult7);

        Table joinResult8 = tableEnv.sqlQuery("select " +
                "s.unique_id	RECORD_ID                                                                            ,"+
                "hospital_name	MED_ORG_NAME                                                                         ,"+
                "hospital_code	MED_ORG_CODE                                                                         ,"+
                "hospital_name	HOSPITAL_NAME                                                                        ,"+
                "hospital_code	HOSPITAL_CODE                                                                        ,"+
                "age_unit	AGE_UNIT                                                                                 ,"+
                "age	AGE                                                                                          ,"+
                "sex	SEX_NAME                                                                                     ,"+
                "sex_code	SEX_CODE                                                                                 ,"+
                "identify_no	ID_CARD_NUMBER                                                                       ,"+
                "patient_typecode	PATIENT_VISIT_CATEGORY_CODE                                                      ,"+
                "patient_type	PATIENT_VISIT_CATEGORY                                                               ,"+
                "case when patient_typecode ='3' then visit_id else '' end INPATIENT_VISIT_FLOW_ID                   ,"+
                "inpatient	INPATIENT_NO                                                                             ,"+
                "case when patient_typecode <> '3' then visit_id else '' end	OUTPATIENT_VISIT_FLOW_ID             ,"+
                "out_patient_id	OUTPATIENT_NO                                                                        ,"+
                "patient_id	PATIENT_ID                                                                               ,"+
                "patient_name	PATIENT_NAME                                                                         ,"+
                "s.domain_id	DOMAIN_ID                                                                            ,"+
                "bed_no	BED_NO                                                                                       ,"+
                "bed_name	BED_NAME                                                                                 ,"+
                "wards_id	WARD_ID                                                                                  ,"+
                "wards_name	WARD_NAME                                                                                ,"+
                "visit_date	VISIT_DATE                                                                               ,"+
                "visit_times	VISIT_TIMES                                                                          ,"+
                "appy_id	APPY_SHEET_NO                                                                            ,"+
                "author_code	AUTHOR_CODE                                                                          ,"+
                "author_name	AUTHOR_NAME                                                                          ,"+
                "custodian_code	CUSTODIAN_CODE                                                                       ,"+
                "custodian_name	CUSTODIAN_NAME                                                                       ,"+
                "anesth_signdate	ANESTH_SIGNDATE                                                                  ,"+
                "anesth_code	ANESTH_CODE                                                                          ,"+
                "anesth_name	ANES_DOCTOR_NAME                                                                     ,"+
                "admission_date	ADMIT_DATETIME                                                                       ,"+
                "dept_code	DEPT_ID                                                                                  ,"+
                "dept_name	DEPT_NAME                                                                                ,"+
                "preoperative_complication	PREOPERATIVE_COMPLICATION                                                ,"+
                "brief_medical_history	BRIEF_MEDICAL_HISTORY                                                        ,"+
                "allergy_history	ALLERGY_HISTORY                                                                  ,"+
                "weight	PATIENT_WEIGHT                                                                               ,"+
                "mental_state_mark	MENTAL_STATE_FLAG                                                                ,"+
                "abo_code	ABO_BLOOD_CODE                                                                           ,"+
                "abo	ABO_BLOOD_NAME                                                                               ,"+
                "rh_code	RH_BLOOD_CODE                                                                            ,"+
                "rh	RH_BLOOD_NAME	                                                                                 ,"+
                "oper_room	OPERATE_ROOM_NO                                                                          ,"+
                "case when diag_type ='术前诊断' then diag_code else '' end	PRE_OPER_DIAG_ICD_CODE                   ,"+
                "case when diag_type ='术前诊断' then diag_name else '' end	PRE_OPER_DIAG_ICD_NAME                   ,"+
                "needing_attention	CONSIDERATIONS                                                                   ,"+
                "ecg_result	CARDIOGRAM_EXAM_RESULT                                                                   ,"+
                "chest_x_findings	CHEST_RAY_EXAM_RESULT                                                            ,"+
                "ct_result	CT_EXAM_RESULT                                                                           ,"+
                "bu_result	B_ULTRAPHONIC_EXAM_RESULT                                                                ,"+
                "mri_result	MRI_EXAM_RESULT                                                                          ,"+
                "pulmonary_exam	LUNG_FUNCTION_EXAM_RESULT                                                            ,"+
                "blood_exam	BLOOD_ROUTINE_EXAM_RESULT                                                                ,"+
                "routine_urine_exam	URINE_ROUTINE_EXAM_RESUTL                                                        ,"+
                "coagulation_test	COAGULATION_EXAM_RESULT                                                          ,"+
                "liver_function_exam	LIVER_FUNCTION_EXAM_RESULT                                                   ,"+
                "blood_gas_analysis	BLOOD_GAS_ANALYSIS_RESULT                                                        ,"+
                "general_exam	GENERAL_EXAM_RESULT                                                                  ,"+
                "auscultation_result	HEART_AUSCULTATE_RESULT                                                      ,"+
                "pulmonary_auscultation	LUNG_AUSCULTATE_RESULT                                                       ,"+
                "ward_id	ROOM_NO                                                                                  ,"+
                "ward_name	ROOM_NAME                                                                                ,"+
                "limb_exam	LIMBS_EXAM_RESULT                                                                        ,"+
                "spinal_exam	SPINE_EXAM_RESULT                                                                    ,"+
                "celiac_exam	STOMACH_EXAM_RESULT                                                                  ,"+
                "tracheal_exam	TRACHEA_EXAM_RESULT                                                                  ,"+
                "dental_exam	TOOTH_EXAM_RESULT                                                                    ,"+
                "appropriate_anes	ANES_INDICATION                                                                  ,"+
                "anesth_code	ANES_DOCTOR_ID                                                                       ,"+
                "anesth_name	ANES_DOCTOR_SIGN_NAME                                                                ,"+
                "effective_time	ELEC_SIGN_DATETIME                                                                   ,"+
                "g.anes_method_code	INTEND_ANES_METHOD_CODE                                                          ,"+
                "g.anes_method_name	INTEND_ANES_METHOD_NAME                                                          ,"+
                "oper_code	OPER_CODE                                                                                ,"+
                "oper_name	OPER_NAME                                                                                ," +
                "s.xds_type as XDS_TYPE," +
                "s.xds_name as XDS_NAME," +
                "s.xds_version as XDS_VERSION," +
                "visit_id   VISIT_ID," +
                "'dwd_mhs_preoperative_visit' as table_name  "+
                "from hdsd00_06_05 s  "+
                "left join 	tdy_list_diag f	on f.unique_id=s.unique_id and f.xds_type=s.xds_type		"+
                "left join 	tdy_list_qanes g	on g.unique_id=s.unique_id and g.xds_type=s.xds_type	"+
                "left join 	tdy_list_qoper a	on a.unique_id=s.unique_id and a.xds_type=s.xds_type	"+
                "");
        tableEnv.createTemporaryView("join_result8",joinResult8);

    }

    private void createSourceTables(StreamTableEnvironment tableEnv) {
        // hdsd00_05_01
        tableEnv.executeSql("create table hdsd00_05_01 (" +
                "pk                       string," +
                "upload_time              string," +
                "status                   string," +
                "empi                     string," +
                "encounter_id             string," +
                "bed_name                 string," +
                "ward_name                string," +
                "wards_id                 string," +
                "patient_id               string," +
                "patient_domain           string," +
                "visit_domain             string," +
                "visit_id                 string," +
                "visit_times              string," +
                "unique_id                string," +
                "xds_type                 string," +
                "xds_name                 string," +
                "effective_time           string," +
                "xds_version              string," +
                "out_patient_id           string," +
                "domain_id                string," +
                "visit_date               string," +
                "inpatient                string," +
                "patient_type             string," +
                "report_id                string," +
                "patient_typecode         string," +
                "appy_id                  string," +
                "specimen_number          string," +
                "telecom                  string," +
                "identify_no              string," +
                "patient_name             string," +
                "sex_code                 string," +
                "sex                      string," +
                "age                      string," +
                "age_unit                 string," +
                "write_time               string," +
                "author_code              string," +
                "author_name              string," +
                "custodian_name           string," +
                "custodian_code           string," +
                "audit_signdate           string," +
                "audit_code               string," +
                "audit_name               string," +
                "check_signdate           string," +
                "technician_code          string," +
                "technician_name          string," +
                "technician_signdate      string," +
                "examiner_code            string," +
                "examiner_name            string," +
                "apply_date               string," +
                "apply_dept_code          string," +
                "apply_dept               string," +
                "apply_agencycode         string," +
                "apply_organ              string," +
                "bed_no                   string," +
                "ward_id                  string," +
                "dept_code                string," +
                "dept_name                string," +
                "wards_name               string," +
                "hospital_code            string," +
                "hospital_name            string," +
                "inspection_method        string," +
                "inspect_type             string," +
                "report_result            string," +
                "report_dept              string," +
                "agency_name              string," +
                "report_remark            string," +
                "inspect_type_code        string," +
                "inspection_item          string" +
                ")" +
                FlinkSourceUtil.getKafkaWith("hdsd00_05_01","App_05_08"));
        // hdsd00_06_03
        tableEnv.executeSql("create table hdsd00_06_03 (" +
                "pk                               string," +
                "upload_time                      string," +
                "status                           string," +
                "empi                             string," +
                "encounter_id                     string," +
                "bed_name                         string," +
                "ward_name                        string," +
                "wards_id                         string," +
                "patient_id                       string," +
                "patient_domain                   string," +
                "visit_domain                     string," +
                "visit_id                         string," +
                "visit_times                      string," +
                "unique_id                        string," +
                "xds_type                         string," +
                "xds_name                         string," +
                "effective_time                   string," +
                "xds_version                      string," +
                "domain_id                        string," +
                "patient_type                     string," +
                "out_patient_id                   string," +
                "patient_typecode                 string," +
                "inpatient                        string," +
                "appy_id                          string," +
                "identify_no                      string," +
                "patient_name                     string," +
                "sex_code                         string," +
                "sex                              string," +
                "age                              string," +
                "age_unit                         string," +
                "write_time                       string," +
                "author_code                      string," +
                "author_name                      string," +
                "custodian_code                   string," +
                "custodian_name                   string," +
                "executor_date                    string," +
                "executor_code                    string," +
                "executor_order                   string," +
                "visit_date                       string," +
                "bed_no                           string," +
                "ward_id                          string," +
                "dept_code                        string," +
                "dept_name                        string," +
                "wards_name                       string," +
                "hospital_code                    string," +
                "hospital_name                    string," +
                "invasive_sign                    string," +
                "allergy_history_mark             string," +
                "allergy_history                  string," +
                "weight                           string," +
                "handling_guidance                string," +
                "order_use_remark                 string," +
                "treatment_plan                   string," +
                "followup_date                    string," +
                "fu_way_code                      string," +
                "fu_way                           string," +
                "followup_recommendcode           string," +
                "followup_recommend               string" +
                ")" +
                FlinkSourceUtil.getKafkaWith("hdsd00_06_03","App_05_08"));
        // hdsd00_06_06
        tableEnv.executeSql("create table hdsd00_06_06 (" +
                "pk                          String,"+
                "upload_time                 String,"+
                "status                      String,"+
                "empi                        String,"+
                "encounter_id                String,"+
                "bed_name                    String,"+
                "ward_name                   String,"+
                "wards_id                    String,"+
                "patient_id                  String,"+
                "patient_domain              String,"+
                "visit_domain                String,"+
                "visit_id                    String,"+
                "visit_times                 String,"+
                "unique_id                   String,"+
                "xds_type                    String,"+
                "xds_name                    String,"+
                "effective_time              String,"+
                "xds_version                 String,"+
                "domain_id                   String,"+
                "patient_type                String,"+
                "out_patient_id              String,"+
                "patient_typecode            String,"+
                "inpatient                   String,"+
                "appy_id                     String,"+
                "identify_no                 String,"+
                "patient_name                String,"+
                "sex_code                    String,"+
                "sex                         String,"+
                "age                         String,"+
                "age_unit                    String,"+
                "write_time                  String,"+
                "author_code                 String,"+
                "author_name                 String,"+
                "custodian_code              String,"+
                "custodian_name              String,"+
                "surgeon_signdate            String,"+
                "visit_date                  String,"+
                "bed_no                      String,"+
                "ward_id                     String,"+
                "dept_code                   String,"+
                "dept_name                   String,"+
                "wards_name                  String,"+
                "hospital_code               String,"+
                "hospital_name               String,"+
                "surgery_history             String,"+
                "oper_code                   String,"+
                "oper_start_date             String,"+
                "oper_end_date               String,"+
                "surgeon_code                String,"+
                "surgeon_name                String,"+
                "assistant_icode             String,"+
                "assistant_i                 String,"+
                "assistant_iicode            String,"+
                "assistant_ii                String,"+
                "instrument_code             String,"+
                "instrument_name             String,"+
                "patrol_code                 String,"+
                "patrol_name                 String,"+
                "oper_name                   String,"+
                "oper_room                   String,"+
                "oper_level_code             String,"+
                "oper_level                  String,"+
                "bleeding                    String,"+
                "blood_volume                String,"+
                "blood_reaction_mark         String,"+
                "anes_method_code            String,"+
                "anes_method_name            String,"+
                "anesth_name                 String,"+
                "pre_medication              String,"+
                "inoper_medical              String,"+
                "infusion_volume             String,"+
                "operation_desc              String,"+
                "target_name                 String,"+
                "intervention                String,"+
                "oper_post_code              String,"+
                "oper_post                   String,"+
                "skin_disinfection           String,"+
                "incision_desc               String,"+
                "drainage_mark               String,"+
                "drainage_material           String,"+
                "drainage_material_num       String,"+
                "placement_site              String"+
                ")" +
                FlinkSourceUtil.getKafkaWith("hdsd00_06_06","App_05_08"));
        // hdsd00_06_05
        tableEnv.executeSql("create table hdsd00_06_05 (" +
                "pk                         String,"+
                "upload_time                String,"+
                "status                     String,"+
                "empi                       String,"+
                "encounter_id               String,"+
                "bed_name                   String,"+
                "ward_name                  String,"+
                "wards_id                   String,"+
                "visit_date                 String,"+
                "patient_id                 String,"+
                "patient_domain             String,"+
                "visit_domain               String,"+
                "visit_id                   String,"+
                "visit_times                String,"+
                "unique_id                  String,"+
                "xds_type                   String,"+
                "xds_name                   String,"+
                "effective_time             String,"+
                "xds_version                String,"+
                "domain_id                  String,"+
                "out_patient_id             String,"+
                "patient_type               String,"+
                "inpatient                  String,"+
                "patient_typecode           String,"+
                "appy_id                    String,"+
                "identify_no                String,"+
                "patient_name               String,"+
                "sex_code                   String,"+
                "sex                        String,"+
                "age                        String,"+
                "age_unit                   String,"+
                "write_time                 String,"+
                "author_code                String,"+
                "author_name                String,"+
                "custodian_code             String,"+
                "custodian_name             String,"+
                "anesth_signdate            String,"+
                "anesth_code                String,"+
                "anesth_name                String,"+
                "admission_date             String,"+
                "bed_no                     String,"+
                "ward_id                    String,"+
                "dept_code                  String,"+
                "dept_name                  String,"+
                "wards_name                 String,"+
                "hospital_code              String,"+
                "hospital_name              String,"+
                "preoperative_complication  String,"+
                "brief_medical_history      String,"+
                "allergy_history            String,"+
                "weight                     String,"+
                "general_exam               String,"+
                "mental_state_mark          String,"+
                "auscultation_result        String,"+
                "pulmonary_auscultation     String,"+
                "limb_exam                  String,"+
                "spinal_exam                String,"+
                "celiac_exam                String,"+
                "tracheal_exam              String,"+
                "dental_exam                String,"+
                "abo_code                   String,"+
                "abo                        String,"+
                "rh_code                    String,"+
                "rh                         String,"+
                "ecg_result                 String,"+
                "chest_x_findings           String,"+
                "ct_result                  String,"+
                "bu_result                  String,"+
                "mri_result                 String,"+
                "pulmonary_exam             String,"+
                "blood_exam                 String,"+
                "routine_urine_exam         String,"+
                "coagulation_test           String,"+
                "liver_function_exam        String,"+
                "blood_gas_analysis         String"+
                ")" +
                FlinkSourceUtil.getKafkaWith("hdsd00_06_05","App_05_08"));
        // tdy_list_qoper
        tableEnv.executeSql("create table tdy_list_qoper (" +
                "pk               String,"+
                "unique_id        String,"+
                "domain_id        String,"+
                "xds_type         String,"+
                "xds_version      String,"+
                "anes_method_code String,"+
                "anes_method_name String,"+
                "contraindication String,"+
                "oper_code        String,"+
                "oper_date        String,"+
                "oper_method      String,"+
                "oper_name        String,"+
                "oper_room        String,"+
                "oper_surgical    String,"+
                "pre_preparation  String"+
                ")"+
                FlinkSourceUtil.getKafkaWith("tdy_list_qoper","App_05_08"));
        // tdy_list_qanes
        tableEnv.executeSql("create table tdy_list_qanes (" +
                "pk                 String,"+
                "unique_id          String,"+
                "domain_id          String,"+
                "xds_type           String,"+
                "xds_version        String,"+
                "anes_method_code   String,"+
                "anes_method_name   String,"+
                "appropriate_anes   String,"+
                "needing_attention  String,"+
                "pre_anes_order     String"+
                ")"+
                FlinkSourceUtil.getKafkaWith("tdy_list_qanes","App_05_08"));

        // tdy_list_diag
        super.readTdyListDiag(tableEnv,"App_05_08");
        // tdy_list_exam
        super.readTdyListExam(tableEnv,"App_05_08");
    }
}
